package com.consumer;
import com.ai.util.ThreadLocalDateUtil;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.consumer.PartitionTopicInfo;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Date;
import java.util.Random;

public class writeFileThread extends Thread {
    private KafkaStream<byte[], byte[]> stream;

    public writeFileThread(KafkaStream<byte[], byte[]> stream) {
        this.stream = stream;
    }
    public void run() {
        try {
            ConsumerIterator<byte[], byte[]> it = stream.iterator();
            int num = 1;
            Number consumedOffset = 0;
            Number fetchedOffset = 0;
            FileWriter fw = null;
            BufferedWriter bw= null;
            StringBuilder curFilename =  new StringBuilder();
            Random ne=new Random();
            while (it.hasNext()) {
                if(num==1){
                    curFilename.append(Thread.currentThread().getName()).append(consumedOffset).append(System.currentTimeMillis());
                    fw = new FileWriter(getFile(curFilename.toString()));
                    bw = new BufferedWriter(fw);
                }
                String content = new String(it.next().message());
                StringBuilder message = new StringBuilder();
                message.append(num).append(";").append(ThreadLocalDateUtil.formatDate(new Date(
                        System.currentTimeMillis()))).append(";").append(content);
                bw.write(message.toString());
                bw.newLine();
                bw.flush();
                if(num%3000==0){
                    PartitionTopicInfo s = it.kafka$consumer$ConsumerIterator$$currentTopicInfo();
                    consumedOffset  = s.kafka$consumer$PartitionTopicInfo$$consumedOffset();
                    fetchedOffset = s.kafka$consumer$PartitionTopicInfo$$fetchedOffset();
                    fw.close();
                    bw.close();
                    //  System.out.println("b  curFilename="+curFilename+" close");
                    fileRename(curFilename.toString());

                    curFilename = new StringBuilder();
                    curFilename.append(Thread.currentThread().getName()).append(consumedOffset).
                            append(System.currentTimeMillis()).append(ne.nextInt(100));
                    fw = new FileWriter(getFile(curFilename.toString()));
                    bw = new BufferedWriter(fw);
                }

                if((fetchedOffset.longValue()-consumedOffset.longValue())<=1000&&consumedOffset.longValue()>0){
                    PartitionTopicInfo s = it.kafka$consumer$ConsumerIterator$$currentTopicInfo();
                    consumedOffset = s.kafka$consumer$PartitionTopicInfo$$consumedOffset();
                    if((fetchedOffset.longValue()-consumedOffset.longValue())==0){
                        fw.close();
                        bw.close();
                        //System.out.println("b  " +fetchedOffset.longValue()+"  "+consumedOffset.longValue());
                        //  System.out.println("b  curFilename="+curFilename+" close");
                        fileRename(curFilename.toString());

                        consumedOffset  = s.kafka$consumer$PartitionTopicInfo$$consumedOffset();
                        fetchedOffset = s.kafka$consumer$PartitionTopicInfo$$fetchedOffset();
                        curFilename = new StringBuilder();
                        curFilename.append(Thread.currentThread().getName()).append(consumedOffset).
                                append(System.currentTimeMillis()).append(ne.nextInt(100));;
                        fw = new FileWriter(getFile(curFilename.toString()));
                        bw = new BufferedWriter(fw);
                    }
                }
                num++;
            }
        } catch (Exception e1) {
            e1.printStackTrace();
        } finally {


        }
    }
    public File getFile(String  filename ) throws IOException{
        File file = null ;

        file = new File("/Users/yinpeng/testlog/"+filename);
        if (!file.exists()) {
            file.createNewFile();
        }
        return file;
    }

    public  void fileRename(String filename) {
        String path = "/Users/yinpeng/testlog/";

        File file1 = new File(path+filename);
        if (file1.exists()) {
            if (file1.isFile()) {
                File file2 = new File(path+filename+"#");
                file1.renameTo(file2);
                //  System.out.println(flag);
            } else {
                System.out.println("非文件");
            }
        } else {
            System.out.println("不存在");
        }
    }



}
